package com.weiyi.weiman.client;

import com.alibaba.fastjson.JSON;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry.*;
import com.alibaba.otter.canal.protocol.Message;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.weiyi.weiman.domain.Const.ComicsConst;
import com.weiyi.weiman.mapper.comics.ComicsCarouselMapper;
import com.weiyi.weiman.mapper.comics.ComicsChapterMapper;
import com.weiyi.weiman.utils.RedisUtils;
import comics.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.Objects;

import static com.baomidou.mybatisplus.core.toolkit.StringUtils.*;

/**
 * canal客户端类，用以监听mysql数据变化
 */
@Slf4j
@Component
public class AppletCanalClient {

    @Value("${canal.host}")
    private String host;
    @Value("${canal.port}")
    private int port;
    @Value("${canal.destination}")
    private String destination;

    @Autowired
    private ComicsChapterMapper chapterMapper;
    @Autowired
    private ComicsCarouselMapper carouselMapper;
    @Autowired
    private RedisUtils redisUtils;

    /**
     * canal入库方法
     */
    public void run() {
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(host, port), destination, "", "");
        try {
            connector.connect();
            //监听的表，    格式为数据库.表名,数据库.表名
            connector.subscribe("weiman1\\..*");
            connector.rollback();

            while (true) {
                Message message = connector.getWithoutAck(100); // 获取指定数量的数据
                long batchId = message.getId();
                if (batchId == -1 || message.getEntries().isEmpty()) {
                    Thread.sleep(1000);
                    continue;
                }
                printEntries(message.getEntries());
                connector.ack(batchId);// 提交确认，消费成功，通知server删除数据
            }
        }catch (Exception e){

        }finally {
            connector.disconnect();
        }
    }

    private void printEntries(List<Entry> entries) throws Exception {
        for (Entry entry : entries) {
            if (entry.getEntryType() != EntryType.ROWDATA) {
                continue;
            }

            RowChange rowChange = RowChange.parseFrom(entry.getStoreValue());

            EventType eventType = rowChange.getEventType();
            System.out.println(String.format("================> binlog[%s:%s] , name[%s,%s] , eventType : %s",
                    entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
                    entry.getHeader().getSchemaName(), entry.getHeader().getTableName(), eventType));

            for (RowData rowData : rowChange.getRowDatasList()) {
                switch (rowChange.getEventType()) {
                    case INSERT:
                        if ("wm_comics_chapter".equals(entry.getHeader().getTableName()))
                            insertChangeChapter(rowData.getAfterColumnsList());
                        if ("wm_comics_carousel".equals(entry.getHeader().getTableName()))
                            insertChangeCarousel(rowData.getAfterColumnsList());
                        if ("wm_category".equals(entry.getHeader().getTableName()))
                            insertChangeCategory(rowData.getAfterColumnsList());
                        if ("wm_comics_rank".equals(entry.getHeader().getTableName()))
                            insertChangeRank(rowData.getAfterColumnsList());
                        break;
                    case UPDATE:
                        if ("wm_comics".equals(entry.getHeader().getTableName()))
                            changeComicsRedis(rowData.getAfterColumnsList());
                        if ("wm_comics_carousel".equals(entry.getHeader().getTableName()))
                            updateChangeCarousel(rowData.getAfterColumnsList());
                        if ("wm_category".equals(entry.getHeader().getTableName()))
                            updateChangeCategory(rowData.getAfterColumnsList());
                        if ("wm_comics_rank".equals(entry.getHeader().getTableName()))
                            updateChangeRank(rowData.getAfterColumnsList());
                        break;
                    case DELETE:
                        System.out.println("DELETE ");
                        printColumns(rowData.getBeforeColumnsList());
                        break;

                    default:
                        break;
                }
            }
        }
    }

    private void printColumns(List<Column> columns) {
        for (Column column : columns) {
            System.out.println(column.getName() + " : " + column.getValue() + " update=" + column.getUpdated());
        }
    }

    private void changeComicsRedis(List<Column> columns) {
        String comicsId = null;
        String comicsKey = ComicsConst.COMICS_REDIS_PRE;
        for (Column column : columns) {
            if (column.getIsKey()) {
                comicsId = column.getValue();
                comicsKey += comicsId;
            }
            if(column.getUpdated()) {
                if ("comics_status".equals(column.getName())) {
                    if ("1".equals(column.getValue())) {
                        // 禁用漫画，将漫画从redis中删除
                        redisUtils.delete(comicsKey);
                        log.info("漫画:{},被禁用,从缓存中删除",comicsId);
                    }
                } else if (redisUtils.hExists(comicsKey, "comicsId")) {
                    redisUtils.hPut(comicsKey, underlineToCamel(column.getName()),column.getValue());
                    log.info("漫画:{},字段{}发生更新,修改值为{}",comicsId,column.getName(),column.getValue());
                }
            }
        }
    }

    /**
     * 处理插入章节的上一章和下一章信息
     * @param columns
     */
    private void insertChangeChapter(List<Column> columns) {
        long chapterId = Long.parseLong(columns.get(0).getValue());
        ComicsChapter chapter = chapterMapper.selectById(chapterId);
        QueryWrapper<ComicsChapter> queryWrapper = new QueryWrapper<>();
        queryWrapper.eq("chapter_num", chapter.getChapterNum() - 1)
                .eq("comics_id", chapter.getComicsId());
        ComicsChapter previousChapter = chapterMapper.selectOne(queryWrapper);
        if (!Objects.isNull(previousChapter)) {
            previousChapter.setNextChapterId(chapterId);
            chapter.setPreviousChapterId(previousChapter.getChapterId());
            chapterMapper.updateById(previousChapter);
            chapterMapper.updateById(chapter);
        }
        queryWrapper = new QueryWrapper<>();
        queryWrapper.eq("chapter_num", chapter.getChapterNum() + 1)
                .eq("comics_id", chapter.getComicsId());
        ComicsChapter nextChapter = chapterMapper.selectOne(queryWrapper);
        if (!Objects.isNull(nextChapter)) {
            nextChapter.setPreviousChapterId(chapterId);
            chapter.setNextChapterId(nextChapter.getChapterId());
            chapterMapper.updateById(nextChapter);
            chapterMapper.updateById(chapter);
        }
        log.info("章节:{}的上下章id设置完成", chapterId);

        //将章节信息加入到redis中
        String chapterListKey = ComicsConst.CHAPTER_LIST_REDIS_PRE + chapter.getComicsId();
        if (redisUtils.hasKey(chapterListKey)) {
            redisUtils.lRightPush(chapterListKey, JSON.toJSONString(chapter));
            log.info("章节:{}信息已经存入缓存中", chapterId);
        }
    }

    /**
     * 轮播图插入数据时保证数据库和缓存读写一致
     * @param columns
     */
    private void insertChangeCarousel(List<Column> columns) {
        if (redisUtils.hasKey(ComicsConst.COMICS_CAROUSEL)) {
            String carouselId = columns.get(0).getValue();
            ComicsCarousel carousel = carouselMapper.selectById(carouselId);
            redisUtils.lRightPush(ComicsConst.COMICS_CAROUSEL, JSON.toJSONString(carousel));
            redisUtils.incrBy(ComicsConst.COMICS_CAROUSEL_NUM, 1);
            log.info("增加轮播图:{},已加入缓存中", carouselId);
        }
    }

    /**
     * 轮播图数据删除或恢复时保证数据库和缓存读写一致
     * @param columns
     */
    private void updateChangeCarousel(List<Column> columns) {
        if (redisUtils.hasKey(ComicsConst.COMICS_CAROUSEL)) {
            ComicsCarousel carousel = new ComicsCarousel();
            boolean deletedFlag = false;
            for (Column column : columns) {
                if (column.getIsKey())
                    carousel.setCarouselId(Long.parseLong(column.getValue()));

                if ("comics_id".equals(column.getName()))
                    carousel.setComicsId(Long.parseLong(column.getValue()));

                if ("carousel_cover".equals(column.getName()))
                    carousel.setCarouselCover(column.getValue());

                if (column.getUpdated() && "carousel_status".equals(column.getName())) {
                    if ("1".equals(column.getValue()))
                        deletedFlag = true;
                    else if ("0".equals(column.getValue()))
                        deletedFlag = false;
                }
            }
            List<String> list = redisUtils.lRange(ComicsConst.COMICS_CAROUSEL, 0, -1);
            for (String carouselJson : list) {
                ComicsCarousel comicsCarousel = JSON.parseObject(carouselJson, ComicsCarousel.class);
                if (comicsCarousel.getCarouselId() == carousel.getCarouselId()) {
                    redisUtils.lRemove(ComicsConst.COMICS_CAROUSEL, 0, carouselJson);
                    redisUtils.set(ComicsConst.COMICS_CAROUSEL_NUM, list.size()+"");
                    break;
                }
            }
            if (!deletedFlag) {
                redisUtils.lRightPush(ComicsConst.COMICS_CAROUSEL, JSON.toJSONString(carousel));
                redisUtils.incrBy(ComicsConst.COMICS_CAROUSEL_NUM, 1);
            }
            log.info("轮播图:{}缓存更新成功", carousel.getCarouselId());
        }
    }

    /**
     * 增加漫画种类时，保证数据库和缓存数据一致
     * @param columns
     */
    private void insertChangeCategory(List<Column> columns) {
        if (redisUtils.hasKey(ComicsConst.CATEGORY_LIST_PRE)) {
            Category category = new Category();
            for (Column column : columns) {
                if (column.getIsKey())
                    category.setCategoryId(Integer.parseInt(column.getValue()));
                if ("category_name".equals(column.getName()))
                    category.setCategoryName(column.getValue());
            }
            redisUtils.lRightPush(ComicsConst.CATEGORY_LIST_PRE, JSON.toJSONString(category));
            log.info("漫画标签:{},加入缓存成功", category.getCategoryId());
        }
    }

    /**
     * 更改漫画种类信息是，保证数据库和缓存数据一致
     * @param columns
     */
    private void updateChangeCategory(List<Column> columns) {
        if (redisUtils.hasKey(ComicsConst.CATEGORY_LIST_PRE)) {
            boolean deleteFlag = false;
            Category category = new Category();
            for (Column column : columns) {
                if (column.getIsKey())
                    category.setCategoryId(Integer.parseInt(column.getValue()));

                if ("category_name".equals(column.getName()))
                    category.setCategoryName(column.getValue());

                if (column.getUpdated() && "deleted".equals(column.getName())) {
                    if ("0".equals(column.getValue()))
                        deleteFlag = false;
                    else if ("1".equals(column.getValue()))
                        deleteFlag = true;
                }
            }
            List<String> categoryList = redisUtils.lRange(ComicsConst.CATEGORY_LIST_PRE, 0, -1);
            for (String categoryJson : categoryList) {
                Category parseObject = JSON.parseObject(categoryJson, Category.class);
                if (parseObject.getCategoryId() == category.getCategoryId()) {
                    redisUtils.lRemove(ComicsConst.CATEGORY_LIST_PRE, 0, categoryJson);
                    break;
                }
            }
            if (!deleteFlag)
                redisUtils.lRightPush(ComicsConst.CATEGORY_LIST_PRE, JSON.toJSONString(category));
            log.info("漫画分类:{}缓存更新成功", category.getCategoryId());
        }
    }

    /**
     * 插入漫画排名标签，保证数据库和缓存数据一致性
     * @param columns
     */
    private void insertChangeRank(List<Column> columns) {
        if (redisUtils.hasKey(ComicsConst.RANK_LIST_PRE)) {
            ComicsRank comicsRank = new ComicsRank();
            for (Column column : columns) {
                if (column.getIsKey())
                    comicsRank.setRankId(Integer.parseInt(column.getValue()));
                if ("rank_name".equals(column.getName()))
                    comicsRank.setRankName(column.getValue());
            }
            redisUtils.lRightPush(ComicsConst.RANK_LIST_PRE, JSON.toJSONString(comicsRank));
            log.info("漫画排名标签:{},已加入缓存", comicsRank.getRankId());
        }
    }

    /**
     * 更改漫画排名标签信息，保证数据库和缓存数据一致性
     * @param columns
     */
    private void updateChangeRank(List<Column> columns) {
        if (redisUtils.hasKey(ComicsConst.RANK_LIST_PRE)) {
            ComicsRank comicsRank = new ComicsRank();
            boolean deletedFlag = false;
            for (Column column : columns) {
                if (column.getIsKey())
                    comicsRank.setRankId(Integer.parseInt(column.getValue()));
                if ("rank_name".equals(column.getName()))
                    comicsRank.setRankName(column.getValue());

                if (column.getUpdated() && "deleted".equals(column.getName())) {
                    if ("0".equals(column.getValue()))
                        deletedFlag = false;
                    else if ("1".equals(column.getValue()))
                        deletedFlag = true;
                }
            }
            List<String> rankList = redisUtils.lRange(ComicsConst.RANK_LIST_PRE, 0, -1);
            for (String rankJson : rankList) {
                ComicsRank rank = JSON.parseObject(rankJson, ComicsRank.class);
                if (rank.getRankId() == comicsRank.getRankId()) {
                    redisUtils.lRemove(ComicsConst.RANK_LIST_PRE, 0, JSON.toJSONString(rank));
                    break;
                }
            }
            if (!deletedFlag)
                redisUtils.lRightPush(ComicsConst.RANK_LIST_PRE, JSON.toJSONString(comicsRank));
            log.info("漫画排名标签:{}缓存更新成功", comicsRank.getRankId());
        }
    }



    private String underlineToCamel(String param){
        if (param==null||"".equals(param.trim())){
            return "";
        }
        int len=param.length();
        StringBuilder sb=new StringBuilder(len);
        for (int i = 0; i < len; i++) {
            char c = Character.toLowerCase(param.charAt(i));
            if (c == UNDERLINE){
                if (++i<len){
                    sb.append(Character.toUpperCase(param.charAt(i)));
                }
            }else{
                sb.append(c);
            }
        }
        return sb.toString();
    }
}

